package com.example.java.kafka.kafkaclients;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.config.TopicConfig;

import java.util.Collections;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;

public class KafkaAdminExample {
    public static void main(String[] args) {
        // 配置 AdminClient
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "43.139.50.31:9092");
        props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
        props.put(AdminClientConfig.RETRIES_CONFIG, 3);
        try (Admin admin = Admin.create(props)) {
            String topicName = "kafka-admin-topic-222";
            // 创建 Topic：3 个分区，2 个副本
            // NewTopic newTopic = new NewTopic(topicName, 3, (short) 2);
            // 创建 Topic：3 个分区，2 个副本 设置 Kafka 主题中每条消息最多保留 1 min
            NewTopic newTopic = new NewTopic(topicName, 3, (short) 2).configs(Collections.singletonMap(TopicConfig.RETENTION_MS_CONFIG, "60000"));
            CreateTopicsResult createResult = admin.createTopics(Collections.singletonList(newTopic));
            try {
                // 等待创建完成
                createResult.all().get();
                System.out.println("主题 '" + topicName + "' 创建成功");
            } catch (ExecutionException e) {
                if (e.getCause() instanceof org.apache.kafka.common.errors.TopicExistsException) {
                    System.out.println("主题已存在，跳过创建");
                } else {
                    throw e;
                }
            }
            // 列出所有主题
            ListTopicsResult listResult = admin.listTopics();
            Set<String> topics = listResult.names().get();
            System.out.println("当前所有主题: " + topics);
            // 查看主题详细信息
            DescribeTopicsResult describeResult = admin.describeTopics(topics);
            describeResult.values().forEach((name, topicFuture) -> {
                try {
                    TopicDescription desc = topicFuture.get();
                    System.out.println("主题 '" + name + "': 分区数=" + desc.partitions().size() + ", 副本数=" + desc.partitions().get(0).replicas().size());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
            // 4. 删除主题（确保 delete.topic.enable=true）
//            DeleteTopicsResult deleteResult = admin.deleteTopics(Collections.singletonList(topicName));
//            deleteResult.all().get();
//            System.out.println("主题 '" + topicName + "' 删除成功");
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }
}